#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <unistd.h>
#include <assert.h>
#include <pthread.h>
#include <signal.h>
#include <cm/thread/thread_pool.h>

/***** BEGIN  线程池工作队列的定义与操作部分 *****/

/* 线程任务链表 */
typedef struct _thread_work_node {
	thread_pool_work work;
	void *arg;
	struct _thread_work_node *next;/* 下一个节点 */
} thread_work_node;

/*
 * @brief	新建一个工作节点，在堆上分配空间，需要外部释放
 * @param	work：工作内容（函数指针）
 * 		arg：需要传递给工作的参数   注意，这些指针需要保证在线程工作期间不会被释放
 * @return	成功返回新建节点的指针，失败返回NULL
 */
static inline thread_work_node* new_work_node(thread_pool_work work, void *arg)
{
	thread_work_node *node = (thread_work_node *) calloc(1,
			sizeof(thread_work_node));
	if (node != NULL) {
		node->work = work;
		node->arg = arg;
		node->next = NULL;
	}
	return node;
}

/*
 * @brief	将工作节点node添加到队列最后面，head是队列的头
 */
static inline void push_work(thread_work_node **head, thread_work_node *node)
{
	thread_work_node *temp = *head;
	if (temp != NULL) {
		while (temp->next != NULL)
			temp = temp->next;
		temp->next = node;
	} else {
		// printf("head is null.\n");
		*head = node;
	}
}

/*
 * @brief	弹出一个节点,head向后移动
 */
static inline thread_work_node * pop_work(thread_work_node **head)
{
	thread_work_node *temp = NULL;
	if (*head != NULL) {
		temp = *head;
		*head = temp->next;
	}

	return temp;
}
/*_____END 线程池工作队列的定义与操作部分结束_____*/

/* 线程池对象 */
typedef struct thread_pool_t {
	pthread_mutex_t threads_lock;/* 控制free和counts的并发访问 */
	int capacity; /* 线程池容量，线程的最大数目 */
	int counts; /* 当前线程数目 */
	int free; /* 当前空闲线程数目 */
	pthread_t *tids;

	thread_work_node *works_queue;
	pthread_mutex_t queue_lock;
	pthread_cond_t queue_ready;
	int queue_size;

	int isdestroy;
} thread_pool_t;

/*
 * @brief	线程池各个线程入口函数
 */
static void *thread_routine(void *arg);

/*
 * @brief	创建一个线程池
 * @param	capacity：线程池的容量
 * @return	成功返回非NULL，失败返回NULL
 */
struct thread_pool_t * thread_pool_create(int capacity)
{
	if (capacity <= 0 || capacity > THREAD_POOL_MAX_CAPACITY)
		return NULL;

	thread_pool_t *pool = NULL;
	pool = (thread_pool_t *) calloc(1, sizeof(thread_pool_t));
	if (pool == NULL)
		return NULL; /* no memory */

	pthread_mutex_init(&(pool->threads_lock), NULL);
	pthread_mutex_init(&(pool->queue_lock), NULL);
	pthread_cond_init(&(pool->queue_ready), NULL);

	pool->tids = (pthread_t *) calloc(capacity, sizeof(pthread_t));
	if (pool->tids == NULL) {
		pthread_mutex_destroy(&(pool->threads_lock));
		pthread_mutex_destroy(&(pool->queue_lock));
		pthread_cond_destroy(&(pool->queue_ready));
		free(pool);
		return NULL;
	}
	pool->capacity = capacity;
	pool->counts = 0;
	pool->free = 0;
	pool->queue_size = 0;
	pool->isdestroy = 0;

	return pool;
}

/*
 * @brief	销毁线程池
 * @param	线程池指针
 * @return	0表示成功，非0表示失败
 */
int thread_pool_destroy(thread_pool_t *pool)
{
	if (pool->isdestroy != 0) {
		return -1;
	}

	pool->isdestroy = 1;

	pthread_cond_broadcast(&(pool->queue_ready));
	int idx;
	for (idx = 0; idx < pool->counts; idx++) {
		int ret = pthread_kill(pool->tids[idx], 0);
		if (ESRCH == ret) {
			continue;
		} else {
			idx--;/* 继续等待停止 */
			sleep(1);
		}
	}

	free(pool->tids);
	pool->tids = NULL;

	while (1) {
		thread_work_node *node;
		node = pop_work(&(pool->works_queue));
		if (node == NULL)
			break;
		free(node);
	}

	pthread_mutex_destroy(&(pool->threads_lock));
	pthread_mutex_destroy(&(pool->queue_lock));
	pthread_cond_destroy(&(pool->queue_ready));
	free(pool);

	return 0;
}

/*
 * @brief	在pool中创建一个游离态的线程
 * @param	pool：线程池对象，通过thread_pool_create创建
 * 		idx：
 * return	成功返回0 失败返回非0
 */
static int thread_create_detach(thread_pool_t *pool, int idx)
{
	int ret = 0;
	pthread_attr_t attr;
	do {
		ret = pthread_attr_init(&attr);
		if (0 != ret) {
			return -1;
		}
		ret = pthread_attr_setdetachstate(&attr,
		PTHREAD_CREATE_DETACHED);
		if (0 != ret) {
			return -1;
		}

		ret = pthread_create(&(pool->tids[idx]), &attr,
				thread_routine, pool);
		if (0 != ret) {
			pthread_attr_destroy(&attr);
			if (EINTR == errno)
				continue;
		} else {
			pthread_attr_destroy(&attr);
		}
	} while (0);

	return ret;
}

/*
 * @brief	保证已经创建的线程存活
 */
static inline int thread_pool_keepalive(thread_pool_t *pool)
{
	int idx = 0, ret = 0;
	for (idx = 0; idx < pool->counts; idx++) {
		ret = pthread_kill(pool->tids[idx], 0);
		if (ESRCH == ret) {
			ret = thread_create_detach(pool, idx);
			if (ret < 0) {
				return -1;
			}
		}
	}

	return 0;
}

/*
 * @brief	添加一个工作到线程池中
 * @param	pool：通过thread_pool_create创建的线程池
 * 		work：工作函数指针
 * 		arg：传递给工作函数指针的参数
 * @return	成功返回0，失败返回非0
 */
int thread_pool_add_work(struct thread_pool_t *pool, thread_pool_work work,
		void *arg)
{
	thread_work_node *node = new_work_node(work, arg);
	if (node == NULL)
		return -1;

	pthread_mutex_lock(&(pool->queue_lock));
	push_work(&(pool->works_queue), node);

	pool->queue_size++;
	pthread_mutex_unlock(&(pool->queue_lock));

	pthread_mutex_lock(&(pool->threads_lock));
	// TODO 循环检查一下所有的线程是不是都还存活
	thread_pool_keepalive(pool);

//	printf(
//			"Add work, capacity = %d, counts = %d, free = %d, queue size = %d\n",
//			pool->capacity, pool->counts, pool->free,
//			pool->queue_size);
	if (pool->free < 1 && pool->counts < pool->capacity) {
		// printf("free threads not enough, create new one.\n");
		thread_create_detach(pool, pool->counts);
		pool->counts++;
		pool->free++;
	} else {
		//printf("here.\n");
	}
	pthread_mutex_unlock(&(pool->threads_lock));

	pthread_cond_signal(&(pool->queue_ready));

	return 0;
}

/*
 * @brief	线程池各个线程入口函数
 * @param	arg：线程池对象
 */
static void *thread_routine(void *arg)
{
	struct thread_pool_t *pool = (struct thread_pool_t*) arg;

	while (1) {
		pthread_mutex_lock(&(pool->queue_lock));
		while ((0 == pool->isdestroy) && (0 == pool->queue_size)) {
			pthread_cond_wait(&(pool->queue_ready),
					&(pool->queue_lock));
		}

		if (pool->isdestroy) { /* 线程退出条件 */
			pthread_mutex_unlock(&(pool->queue_lock));
			pthread_exit(NULL);
		}

		thread_work_node *work = NULL;
		work = pop_work(&(pool->works_queue));
		if (work)
			pool->queue_size--;

		pthread_mutex_unlock(&(pool->queue_lock));
		/* 执行队列中的任务 */
		if (work) {
			pthread_mutex_lock(&(pool->threads_lock));
			pool->free--;
			pthread_mutex_unlock(&(pool->threads_lock));
			(*(work->work))(work->arg);
			free(work);
			work = NULL;
			pthread_mutex_lock(&(pool->threads_lock));
			pool->free++;
//			printf(
//					"One work done, capacity = %d, counts = %d, free = %d, queue size = %d\n",
//					pool->capacity, pool->counts,
//					pool->free, pool->queue_size);
			pthread_mutex_unlock(&(pool->threads_lock));
		}
	}

	return 0;
}

/* DEMO */

//void* myprocess(void *arg)
//{
//	fprintf(stdout, "[%s][%d] threadid:%d arg:%d\n", __FILE__,
//	__LINE__, pthread_self(), *(int*) arg);
//	sleep(5);
//	return NULL;
//}
//
//int main(void)
//{
//	struct thread_pool_t *pool = NULL;
//	int array[32] = { 0 };
//
//	pool = thread_pool_create(4);
//	if (pool == NULL) {
//		return -1;
//	}
//
//	int idx;
//	for (idx = 0; idx < 32; idx++) {
//		array[idx] = idx;
//		sleep(1);
//		thread_pool_add_work(pool, myprocess, &array[idx]);
//	}
//
//	while(1) {
//		sleep(1);
//	}
//	// thread_pool_destroy(pool);
//	// pool = NULL;
//	return 0;
//}
